package saramsx

import (
	"encoding/json"
	"gitee.com/chicken-c/go-learn/webook/pkg/logger"
	"github.com/IBM/sarama"
)

type Handler[T any] struct {
	l  logger.Logger
	fn func(mes *sarama.ConsumerMessage, event T) error
}

func NewHandler[T any](l logger.Logger, fn func(mes *sarama.ConsumerMessage, event T) error) *Handler[T] {
	return &Handler[T]{l: l, fn: fn}
}

func (h *Handler[T]) Setup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h *Handler[T]) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h *Handler[T]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	ch := claim.Messages()
	for mes := range ch {
		var val T
		err := json.Unmarshal(mes.Value, &val)
		if err != nil {
			h.l.Error("反序列化失败", logger.Field{
				Key: "Topic",
				Val: mes.Topic,
			}, logger.Field{
				Key: "Partition",
				Val: mes.Partition,
			}, logger.Field{
				Key: "offset",
				Val: mes.Offset,
			})
		}

		//实际的业务处理函数
		err = h.fn(mes, val)
		if err != nil {
			h.l.Error("消费消息失败", logger.Field{
				Key: "error",
				Val: err,
			})
		}
		session.MarkMessage(mes, "")
	}
	return nil
}
